package com.shujia.flink.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class Demo01FileSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1、读取文件，老版本读文件的方式
        DataStream<String> lineDS = env.readTextFile("flink/data/words.txt");

//        lineDS.print();

        // 2、使用FileSource进行文件读取
        /**
         * 新版读取文件的方式：
         * 1、指定一个格式的方式
         * 2、指定一个路径
         * 3、通过monitorContinuously来实现对目录的实时监控，将读文件转换无界流进行处理
         */
        FileSource<String> fileSource = FileSource.forRecordStreamFormat(
                        new TextLineInputFormat()// 指定读取文件的格式
                        , new Path("flink/data/word/")
                )
                .monitorContinuously(Duration.ofMillis(5)) // 指定时间间隔监控目录的变化
                .build();

        // 使用FileSource
        DataStream<String> wordDS2 = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "fileSource");

        wordDS2.print();

        env.execute();


    }
}
